package com.ipinyou.hb2hf

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import com.ipinyou.hbase.model.MUser
import com.ipinyou.hbase.service.impl.HBaseConnectionServiceImpl
import org.slf4j.LoggerFactory
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.hadoop.hbase.util.Bytes
import scala.collection.mutable.ArrayBuffer
import org.apache.hadoop.hbase.client.Scan

object hb2hf {

  val a = 1.0
  val b = 2.0

  // 0 测试, 1 正式
  var flag = "0"

  var output = "file:///Users/miaoyujia/tmp/hb2hdfs"
  // log
  val log = LoggerFactory.getLogger(this.getClass)

  def main(args: Array[String]): Unit = {

    if (args.length > 0) {
      flag = args(0)
      output = args(1)
    }

    val config = new SparkConf().setAppName("mobile.hb2hdfs")
    val sc = new SparkContext(config)

    val conf = HBaseConfiguration.create()
    val scan = new Scan()
    scan.setCacheBlocks(false)
    conf.set(TableInputFormat.INPUT_TABLE, "mobileprofile")
    conf.set(TableInputFormat.SCAN_MAXVERSIONS, "30")
    conf.set(TableInputFormat.SCAN_CACHEBLOCKS, "false")

    val hBaseRDD = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat],
      classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],
      classOf[org.apache.hadoop.hbase.client.Result])

    println("hbase count: " + hBaseRDD.count())
    println("=====================================================")

    //    //遍历输出
    val resRdd = hBaseRDD.map {
      case (_, result) =>
        //        for (kv <- kvs) yield List(new String(kv.getRow), kv.getTimestamp, new String(kv.getValue))
        val kvs = result.raw
        val list = new ArrayBuffer[Float]()
        val rowkey = new String(result.getRow)
        for (kv <- kvs) {
          list.append(new String(kv.getValue).toFloat)
        }
        (rowkey, list)
    }.flatMap {
      case (k, v) =>
        val score = a * v.take(7).sum + b * v.sum
        val karr = k.split("""\|""")
        val sscore = score.toString
        if (sscore.equals("Infinity")) {
          List()
        } else {
          List(List(karr(2), karr(1), score.toString).mkString("\t"))
        }
    }

    resRdd.repartition(2).saveAsTextFile(output)
  }
}